package com.flink.partition;

import com.flink.entity.User;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.LocalDate;

/**
 * 描述:
 * 物理分区Api
 *
 * @author yanzhengwu
 * @create 2022-07-22 8:28
 */
public class PartitionData {
    public static void main(String[] args) throws Exception {
        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        env.setParallelism(1);

        DataStreamSource<User> stream = env.fromElements(
                new User("用户2", "家电", LocalDate.of(2022, 6, 2), 12, 10000L),
                new User("用户1", "零食", LocalDate.of(2022, 1, 1), 11, 10000L),
                new User("用户2", "洗发水", LocalDate.of(2022, 6, 2), 13, 10000L),
                new User("用户3", "椅子", LocalDate.of(2022, 6, 3), 14, 10000L),
                new User("用户3", "手机", LocalDate.of(2022, 6, 3), 15, 10000L),
                new User("用户1", "电脑", LocalDate.of(2022, 6, 2), 13, 10000L),
                new User("用户1", "手机", LocalDate.of(2022, 6, 16), 16, 10000L),
                new User("用户2", "洗发水", LocalDate.of(2022, 6, 2), 13, 60000L)
        );

        //1、随机分区  可以理解为cpu资源谁抢到就是谁的
//        stream.shuffle().print("shuffle->").setParallelism(4);


        //2、轮训分区 TODO 如果前后并行度不同，flink默认就是轮训分区
//        stream.rebalance().print("rebalance->").setParallelism(4);

        //3、重缩放分区rescale TODO
        /*env.addSource(new RichParallelSourceFunction<Integer>() {
            @Override
            public void run(SourceContext<Integer> ctx) throws Exception {
                for (int i = 1; i <= 8; i++) {
                    //并行两个任务通过任务号进行区分输出哪些数值
                    if (i % 2 == getRuntimeContext().getIndexOfThisSubtask()) {
                        ctx.collect(i);
                    }
                }
            }
            @Override
            public void cancel() {
            }
        })
                .setParallelism(2)
                .rescale()
                .print("rebalance->")
                .setParallelism(4);*/


        //4、广播分区，会将数据向所有分区发送
//        stream.broadcast().print().setParallelism(4);
        //5、全局分区,极端操作，将所有数据全部强行塞进同一个分区进行计算
        //造成严重的数据倾斜，所有并行指令全部失效，都变为1
//        stream.global().print().setParallelism(4);

        //6、自定义分区操作
        DataStreamSource<Integer> intStream = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8);
        intStream.partitionCustom(new Partitioner<Integer>() {
            //重写分区策略
            @Override
            public int partition(Integer key, int numPartitions) {
                return key % 2;
            }
        }, new KeySelector<Integer, Integer>() {
            //重写getkey策略
            @Override
            public Integer getKey(Integer value) throws Exception {
                return value;
            }
        }).print().setParallelism(4);

        env.execute();

    }
}
